/**
 * <b>Copyright (c)</b> 2014青岛儒商信息技术有限公司-版权所有<br/> 
 */
package com.rshang.ip;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.http.HttpResponse;
import org.apache.http.ParseException;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.util.EntityUtils;

import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.jfinal.plugin.activerecord.Db;
import com.jfinal.plugin.activerecord.IAtom;
import com.jfinal.plugin.activerecord.Record;

/**
 * <b>项目名：</b>儒商软件开发平台<br/>
 * <b>包名：</b>com.github.jarod.qqwry.trans<br/>
 * <b>文件名：</b>QQwryAtom.java<br/>
 * <b>功能简介：</b>简要说明类功能<br/>
 * <b>作者：</b>zhangjun<br/>
 * <b>日期：</b>2014年8月13日-上午10:38:22<br/>
 */
public class IpAtomWithAsync implements IAtom {

	private static Log logger = LogFactory.getLog(IpAtomWithAsync.class);
	private Map ipMap;
	private String czr;
	private final String URL = "http://ip.taobao.com/service/getIpInfo.php?ip=";
	private int batchSize = 2;

	public IpAtomWithAsync(Map ipMap,String czr) {
		super();
		this.ipMap = ipMap;
		this.czr = czr;
	}

	/* (non-Javadoc)
	 * @see com.jfinal.plugin.activerecord.IAtom#run()
	 */
	@Override
	public boolean run() throws SQLException {
		boolean success = true;
		//logger.info("ipMap:"+ipMap.toString());
		if(ipMap!=null&&!ipMap.isEmpty())
		{
			String[] begin_arr = ((String)ipMap.get("begin")).split("\\.");
			String[] end_arr = ((String)ipMap.get("end")).split("\\.");
			String beginIp = begin_arr[0];
			String endIp = end_arr[0];
		
            
			//CloseableHttpClient httpClient = HttpClients.createDefault();
			HttpGet httpGet = null;
			CloseableHttpResponse response = null;
			
			RequestConfig requestConfig = RequestConfig.custom().setSocketTimeout(5000).setConnectTimeout(5000).build();//设置请求和传输超时时间
			CloseableHttpAsyncClient httpClient = HttpAsyncClients.custom().setDefaultRequestConfig(requestConfig).build();
			
			httpClient.start();
			final ObjectMapper mapper = new ObjectMapper(); // can reuse, share globally
			HttpGet[] requests = new HttpGet[batchSize];
			int req_count = 0;
			for (int a = Integer.parseInt(beginIp); a <= Integer.parseInt(endIp); a++) {
				for (int b = 0; b <= 255; b++) {
					for (int c = 0; c <= 255; c++) {
						for (int d = 0; d <= 255; d++) {
							final String ip = String.format("%d.%d.%d.%d", a, b,
									c, d);
							logger.info("req_count:"+req_count+" IP:"+ip);
							//httpGet = new HttpGet(URL+ip);
							requests[req_count] = new HttpGet(URL+ip);
							if(req_count==(batchSize-1)||a==Integer.parseInt(endIp))
							{
								//logger.info("requests.length:"+requests.length);
								final CountDownLatch latch = new CountDownLatch(requests.length);
								final ArrayList sqlList = new ArrayList();
								for(final HttpGet request:requests)
								{
									//logger.info("requests.ip:"+request.toString());
									httpClient.execute(request, new FutureCallback(){
										
					                    public void failed(final Exception ex) {
					                        latch.countDown();
					                        Db.save("ip_data_err", new Record().set("ip_address",ip).set("lrr",czr));
					                        //System.out.println(request.getRequestLine() + "->" + ex);
					                    }

					                    public void cancelled() {
					                        latch.countDown();
					                        Db.save("ip_data_err", new Record().set("ip_address",ip).set("lrr",czr));
					                        //System.out.println(request.getRequestLine() + " cancelled");
					                    }

										@Override
										public void completed(Object arg0) {
											final HttpResponse response = (HttpResponse)arg0;
											latch.countDown();
											int status=response.getStatusLine().getStatusCode();
											
											if(status==200)
											{
												IpResult ipData = null;
												try {
													//logger.info("response:"+EntityUtils.toString(response.getEntity()));
													ipData = mapper.readValue(EntityUtils.toString(response.getEntity()).getBytes("UTF-8"), IpResult.class);
												} catch (JsonParseException e) {
													// TODO Auto-generated catch block
													e.printStackTrace();
												} catch (JsonMappingException e) {
													// TODO Auto-generated catch block
													e.printStackTrace();
												} catch (ParseException e) {
													// TODO Auto-generated catch block
													e.printStackTrace();
												} catch (IOException e) {
													// TODO Auto-generated catch block
													e.printStackTrace();
												}
												//logger.info("ipdata:"+ipData.toString());
												//System.out.println(ipData.getCode()+":::"+(ipData.getCode().equals("0")));
												if(ipData!=null&&(ipData.getCode().equals("0")))//code等于0表示成功
												{
													String sql = null;
													try {
														sql = genInsertObject(ipData.getData(),czr);
													} catch (UnsupportedEncodingException e1) {
														// TODO Auto-generated catch block
														e1.printStackTrace();
													}
													logger.info(sql);
													sqlList.add(sql);
													//logger.info("sqlList.size:"+sqlList.size());
													
													if(sqlList.size()==batchSize)
													{
														int[] insertCount =  Db.batch((java.util.List<String>) sqlList, batchSize);
														if(insertCount.length>0)
														{
															logger.info("新插入记录数:"+insertCount.length);
														}
														sqlList.clear();
													}
													
												}
											}
					                        //System.out.println("completed->" + response.getStatusLine());
										}
									});
								}
								try {
									latch.await();
								} catch (InterruptedException e) {
									// TODO Auto-generated catch block
									e.printStackTrace();
								}
								req_count=0;
								continue;
							}
							req_count++;
							
						}
					}
				}
			}
			try {
				httpClient.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		return success; 
	}

	/**
	 * Record对象生成方法<br/>
	 * @param obj		IP对象
	 * @param czr		操作人
	 * @return
	 * 		return Record
	 * @throws UnsupportedEncodingException 
	 * @since  1.0.0
	 * 
	 */
	private String genInsertObject(Ip ip,String czr) throws UnsupportedEncodingException
	{
		/*Record record = new Record();
		record.set("IP_ADDRESS", ip.getIp());
		record.set("country_id", ip.getCountry_id());
		record.set("country_name", ip.getCountry());
		record.set("region_id", ip.getRegion_id());
		record.set("region_name", ip.getCountry());
		record.set("county_id", ip.getCounty_id());
		record.set("county_name", ip.getCounty());
		record.set("city_id", ip.getCity_id());
		record.set("city_name", ip.getCity());
		record.set("area_id", ip.getArea_id());
		record.set("area_name", ip.getArea());
		record.set("isp_id", ip.getIsp_id());
		record.set("isp_name", ip.getIsp());*/
		 return "INSERT INTO IP_DATA (IP_ADDRESS, COUNTRY_ID, COUNTRY_NAME, AREA_ID, AREA_NAME, REGION_ID, REGION_NAME, COUNTY_ID, COUNTY_NAME, CITY_ID, CITY_NAME, ISP_ID, ISP_NAME, LRR) VALUES('"+
		ip.getIp()+"', '"+ip.getCountry_id()+"', '"+unicode2String(ip.getCountry())+"', '"+ip.getArea_id()+"', '"+ip.getArea()+"', '"+ip.getRegion_id()+"', '"+ip.getRegion()+"', '"+ip.getCounty_id()+"', '"+ip.getCounty()+"', '"+ip.getCity_id()+"', '"+ip.getCity()+"', '"+ip.getIsp_id()+"', '"+ip.getIsp()+"', '"+czr+"')";
	}
	/**
	 * unicode转换字符串方法<br/>
	 * (这里描述这个方法适用条件 可选)<br/>
	 * @param unicodeStr		unicode字符串
	 * @return
	 * 	return  String
	 * @throws UnsupportedEncodingException 
	 * @since  1.0.0
	 * 
	 */
	private  String unicode2String(String unicodeStr) throws UnsupportedEncodingException{  
	    return unicodeStr;
		//return new String(new String(unicodeStr.getBytes(),"UTF-8").getBytes(),"GBK");
	 }
}
